package olegroshka.src.com.bofa.olegroshka.datasource;

import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;



public class SequentialJoinDataSource implements DataSource<Object[]> {

	private final static Logger LOG = Logger.getLogger(SequentialJoinDataSource.class.getName());

	private final DataSource<Object[]> leftDataSource;
	private final DataSource<Object[]> rightDataSource;
	private final int leftColumn; 
	private final int rightColumn; 
	private final int leftRecordLength; 
	private final int rightRecordLength; 
	private final int joinedRecordLength; 

	private final Map<Object, Object[]> leftMap = 
			new LinkedHashMap<Object, Object[]>(1000); 
	
	private final Map<Object, Object[]> resultMap = 
			new LinkedHashMap<Object, Object[]>(1000); 
	
	public SequentialJoinDataSource(
			DataSource<Object[]> leftDataSource,
			DataSource<Object[]> rightDataSource,
			int leftColumn, int rightColumn, 
			int leftRecordSize, int rightRecordSize) {
		this.leftDataSource = leftDataSource;
		this.rightDataSource = rightDataSource;
		this.leftColumn = leftColumn;
		this.rightColumn = rightColumn;
		this.leftRecordLength = leftRecordSize;
		this.rightRecordLength = rightRecordSize;
		this.joinedRecordLength = leftRecordSize + rightRecordSize;
		
	}

	public Iterator<Object[]> iterator() {
		return resultMap.values().iterator();
	}

	private void joinRows() {
		Iterator<Object[]> iterator = leftDataSource.iterator();
		while (iterator.hasNext()) {
			Object[] leftRecord = (Object[]) iterator.next();
			Object key = leftRecord[leftColumn];
			leftMap.put(key, leftRecord);
		}
		iterator = rightDataSource.iterator();
		while (iterator.hasNext()) {
			Object[] rightRecord = (Object[]) iterator.next();
			Object key = rightRecord[rightColumn];
			Object[] leftRecord = leftMap.get(key);
			if( leftRecord != null ) {
				Object[] joinedData = new Object[joinedRecordLength];   
				System.arraycopy(leftRecord, 0, joinedData, 0, leftRecordLength);
				System.arraycopy(rightRecord, 0, joinedData, leftRecordLength, rightRecordLength);
				resultMap.put(key, joinedData);
				leftMap.remove(key);
			} else {
				LOG.log(Level.INFO, "Can not match left record to the right : " + 
			            Arrays.toString(rightRecord) + ", key=" + key);
				System.out.println("Can not match left record to the right : " + 
			            Arrays.toString(rightRecord) + ", key=" + key);
			} 
		}
		
		//handle unmatched left records
		Set<Entry<Object, Object[]>> entrySet = leftMap.entrySet();
		for (Entry<Object, Object[]> entry : entrySet) {
			Object[] joinedData = new Object[joinedRecordLength];   
			System.arraycopy(entry.getValue(), 0, joinedData, 0, leftRecordLength);
			Object key = entry.getKey();
			resultMap.put(key, joinedData);
		}
	}

	public void begin() {
		leftDataSource.begin();
		rightDataSource.begin();
		joinRows();
	}

	public void end() {
		leftDataSource.end();
		rightDataSource.end();
		
		resultMap.clear();
		leftMap.clear();
	}
}
